/*
 * BaseAstyanaxDao
 * 
 * Project: HR Boss
 * 
 * Copyright 2012 by Daviteq Co.
 * 10th Floor, TMA Tower, QTSC
 * All rights reserved.
 *
 * This software is the confidential and proprietary information
 * of Glandore Systems. ("Confidential Information"). You
 * shall not disclose such Confidential Information and shall
 * use it only in accordance with the terms of the license
 * agreement you entered into with Glandore Systems.
 */
package com.daviteq.fw.dao.core;

import java.util.Map;

import javax.annotation.Nullable;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.daviteq.fw.utils.Constants;
import com.google.common.base.Function;
import com.netflix.astyanax.AstyanaxConfiguration;
import com.netflix.astyanax.AstyanaxContext;
import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.NodeDiscoveryType;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.connectionpool.impl.ConnectionPoolConfigurationImpl;
import com.netflix.astyanax.connectionpool.impl.CountingConnectionPoolMonitor;
import com.netflix.astyanax.impl.AstyanaxConfigurationImpl;
import com.netflix.astyanax.model.Column;
import com.netflix.astyanax.model.ColumnFamily;
import com.netflix.astyanax.model.ColumnList;
import com.netflix.astyanax.model.Row;
import com.netflix.astyanax.query.RowQuery;
import com.netflix.astyanax.recipes.reader.AllRowsReader;
import com.netflix.astyanax.serializers.StringSerializer;
import com.netflix.astyanax.thrift.ThriftFamilyFactory;
import com.netflix.astyanax.util.RangeBuilder;

/**
 * Basic abstract Data Access class
 * 
 * @author DINH HONG CHAU
 * 
 */
public abstract class BaseAstyanaxDao {
    private static final Logger LOG = LoggerFactory.getLogger(BaseAstyanaxDao.class);
    private static final String KEYSPACE = Constants.SYSTEM_KEYSPACE;
    private static final String HOST = Constants.SYSTEM_HOST;
    private Keyspace keyspace;
    private AstyanaxContext<Keyspace> astyanaxContext;

    /**
     * Initialize Astyanax context, open Cassandra connection with specified configuration.
     */
    public void init() {
        try {
            AstyanaxConfiguration config = new AstyanaxConfigurationImpl().setDiscoveryType(NodeDiscoveryType.NONE)
                    .setCqlVersion(Constants.CQL_VERSION).setTargetCassandraVersion(Constants.CASSANDRA_VERSION);
            this.astyanaxContext = new AstyanaxContext.Builder().forCluster(Constants.SYSTEM_CLUSTER).forKeyspace(KEYSPACE)
                    .withAstyanaxConfiguration(config)
                    .withConnectionPoolConfiguration(new ConnectionPoolConfigurationImpl("MyConnectionPool").setMaxConnsPerHost(1).setSeeds(HOST))
                    .withConnectionPoolMonitor(new CountingConnectionPoolMonitor()).buildKeyspace(ThriftFamilyFactory.getInstance());
            this.astyanaxContext.start();
            this.keyspace = this.astyanaxContext.getClient();

            // test the connection
            this.keyspace.describeKeyspace();
        } catch (Throwable e) {
            LOG.error("Could not connect to Cassandra Server: ", e);
            System.exit(-1);
        }
    }

    /**
     * Shut down Astyanax context.
     */
    public void cleanup() {
        this.astyanaxContext.shutdown();
    }

    /**
     * Insert set of column-value to a row by specified row key.
     * 
     * @param rowKey
     *            Key used to fetch the row
     * @param columns
     *            Map contains column-value entries to be inserted/updated
     * @throws ConnectionException
     */
    public void insert(String rowKey, Map<String, String> columns) throws ConnectionException {
        MutationBatch mutation = this.getKeyspace().prepareMutationBatch();
        for (Map.Entry<String, String> entry : columns.entrySet()) {
            mutation.withRow(getColumnFamily(), rowKey).putColumn(entry.getKey(), entry.getValue(), null);
        }
        mutation.execute();
        LOG.debug("Wrote child [" + rowKey + "]");
    }

    /**
     * Insert new row values to columns of specified row using CQL 3
     * 
     * @throws ConnectionException
     */
    public void writeByCQL() throws ConnectionException {
        this.getKeyspace().prepareQuery(getColumnFamily())
                .withCql("INSERT INTO children(childId, firstname, lastname) values ('the.devil', 'the', 'devil')").execute();
    }

    public ColumnList<String> readWholeColumns(String rowKey) throws ConnectionException {
        return readSlicedColumns(rowKey, null, null);
    }

    public ColumnList<String> readSlicedColumns(String rowKey, String startColumn, String endColumn) throws ConnectionException {
        RowQuery<String, String> rowQuery = keyspace.prepareQuery(getColumnFamily()).getKey(rowKey);
        if (StringUtils.isNotBlank(startColumn) && StringUtils.isNotBlank(endColumn)) {
            rowQuery = rowQuery.withColumnRange(new RangeBuilder().setStart(startColumn).setEnd(endColumn).setLimit(100).build());
        }
        return rowQuery.execute().getResult();
    }

    public ColumnList<String> readColumnSet(String rowKey, String... columns) throws ConnectionException {
        return getKeyspace().prepareQuery(getColumnFamily()).getKey(rowKey).withColumnSlice(columns).execute().getResult();
    }

    public String readSingleColumnValue(String rowKey, String columnName) throws ConnectionException {
        Column<String> column = getKeyspace().prepareQuery(getColumnFamily()).getRow(rowKey).getColumn(columnName).execute().getResult();
        return column.getStringValue();
    }

    public void readAllRowsInColumnFamily() throws Exception {
        new AllRowsReader.Builder<String, String>(getKeyspace(), getColumnFamily()).forEachRow(new Function<Row<String, String>, Boolean>() {
            @Override
            @Nullable
            public Boolean apply(@Nullable Row<String, String> input) {
                // TODO Auto-generated method stub
                return null;
            }
        }).build().call();
    }

    public void readAllRowKeysInColumnFamily() throws Exception {
        new AllRowsReader.Builder<String, String>(getKeyspace(), getColumnFamily()).withColumnRange(null, null, false, 0)
            .forEachRow(new Function<Row<String, String>, Boolean>() {
            @Override
            public Boolean apply(@Nullable Row<String, String> row) {
                // Process the row here ...
                return true;
            }
        }).build().call();
    }

    public int getNumberOfColumn(String rowKey) throws ConnectionException {
        return getKeyspace().prepareQuery(getColumnFamily()).getKey(rowKey).getCount().execute().getResult();
    }

    public void delete(String rowKey) throws ConnectionException {
        MutationBatch mutation = this.getKeyspace().prepareMutationBatch();
        mutation.withRow(getColumnFamily(), rowKey).delete();
        mutation.execute();
    }

    public void update(String rowKey, Map<String, String> columns) {

    }

    public void update(String rowKey, String columnName, byte[] value) {

    }

    public Keyspace getKeyspace() {
        return keyspace;
    }

    public abstract String getColumnFamilyOrTableName();

    public ColumnFamily<String, String> getColumnFamily() {
        return new ColumnFamily<String, String>(getColumnFamilyOrTableName(), StringSerializer.get(), StringSerializer.get());
    }
}
